package com.xfsg.core.kafka.v3;

import com.alibaba.fastjson.JSON;
import com.xfsg.core.kafka.KafkaHeaderUtil;
import com.yomahub.tlog.constant.TLogConstants;
import com.yomahub.tlog.context.SpanIdGenerator;
import com.yomahub.tlog.context.TLogContext;
import com.yomahub.tlog.utils.LocalhostUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author ZhangYi
 * @version 1.0.0
 * @Description:
 * @date 2022/3/29
 */
@Slf4j
public class TracingProducerInterceptor implements ProducerInterceptor<String,String>  {

    public static String serviceId = "";

    @Override
    public ProducerRecord onSend(ProducerRecord<String,String> producerRecord) {
        String traceId = TLogContext.getTraceId();
        if (StringUtils.isNotBlank(traceId)) {
            KafkaHeaderUtil.replaceHeader(producerRecord.headers(), TLogConstants.TLOG_TRACE_KEY, traceId);
            KafkaHeaderUtil.replaceHeader(producerRecord.headers(), TLogConstants.PRE_IVK_APP_KEY, serviceId);
            KafkaHeaderUtil.replaceHeader(producerRecord.headers(), TLogConstants.PRE_IVK_APP_HOST, LocalhostUtil.getHostName());
            KafkaHeaderUtil.replaceHeader(producerRecord.headers(), TLogConstants.PRE_IP_KEY, LocalhostUtil.getHostIp());
            KafkaHeaderUtil.replaceHeader(producerRecord.headers(), TLogConstants.TLOG_SPANID_KEY, SpanIdGenerator.generateNextSpanId());
        }
        return producerRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {
        if(map.containsKey("client.id")){
            serviceId = map.get("client.id").toString();
            log.info("init kafka serviceId:{}",serviceId);
        }
    }
}
